package com.ethan.androidroom;

import android.os.Bundle;

import androidx.appcompat.app.AppCompatActivity;

import java.util.ArrayList;
import java.util.List;

import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableEmitter;
import io.reactivex.rxjava3.core.ObservableOnSubscribe;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.schedulers.Schedulers;

public class RxActivity extends AppCompatActivity {

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_rxmain);

//        //create操作符 onConplete和onE
//        Observable.create((ObservableOnSubscribe<String>) emitter -> {
//            emitter.onNext("12121");
//            emitter.onComplete();
//            emitter.onError(new NullPointerException());
//
//        }).subscribe(s -> {
//            System.out.println("----"+s);
//        },e->{
//            System.out.println("----"+e);
//        });
//        //just操作符
//        Observable.just("1","2").subscribe(s->{
//            System.out.println("just:"+s);
//        });
        subScribeOn();
    }

    public static void main(String[] args) {
        buffer();
    }
    //create操作符
    private void create(){
        Observable.create((ObservableOnSubscribe<String>) emitter -> {

        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onNext(@NonNull String s) {

            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }
    //just操作符
    private static void  just(){
        Observable.just("1","2").subscribe(s->{
            System.out.println("just:"+s);
        });
    }
    //just操作符
    private static void from(){
        Observable.fromArray("1","2","3").subscribe(s->{
            System.out.println("just:"+s);
        });
        List<String> list  = new ArrayList<>();
        list.add("11");
        list.add("22");
        Observable.fromIterable(list).subscribe(s->{
            System.out.println("just:"+s);
        });
    }
    private static void map(){
        Observable.just("1").map(v->{
            return v+"121";
        }).subscribe(s->{
            System.out.println(s);
        });
    }
    private static void flatMap(){
        //flastmap无序的 concatMap(有序)
        Observable.just("1").flatMap(v->{
            return Observable.just("121");
        }).subscribe(s->{
            System.out.println(s);
        });
    }
    private static void buffer(){
        Observable.just("1","2,","3","4","5","6","7").buffer(3).subscribe(v->{
            System.out.println(v);
        });
    }
    public static void subScribeOn(){
        Observable.create(new ObservableOnSubscribe<Object>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Object> emitter) throws Throwable {
                emitter.onNext("12121");
                System.out.println("----"+Thread.currentThread());
            }
        }).subscribeOn(Schedulers.io()).map(v->{
            System.out.println("----"+Thread.currentThread());
            return "1212";
        }).observeOn(AndroidSchedulers.mainThread()).
                map(v->{
                    System.out.println("----"+Thread.currentThread());
                    return "123432";
                }).subscribe(v->{
            System.out.println("----"+Thread.currentThread());
        });
    }


}